package com.media.xxljob;

import com.media.mapper.MediaProcessMapper;
import com.media.po.MediaProcess;
import com.media.service.MediaFilesService;
import com.media.service.MediaProcessService;
import com.utils.Mp4VideoUtil;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class MediaSchedule {

    @Autowired
    private MediaProcessService mediaProcessService;

    @Autowired
    private MediaProcessMapper mediaProcessMapper;

    @Autowired
    private MediaFilesService mediaFilesService;

    @Value("${videoprocess.ffmpegpath}")
    private String ffmpegpath;


    @XxlJob("MediaJobHandler")
    public void MediaJobHandler() throws Exception {
        //分片序号
        int shardIndex = XxlJobHelper.getShardIndex();
        //分片总数
        int shardTotal = XxlJobHelper.getShardTotal();

        log.info("分片序号：{}，分片总数：{}", shardIndex, shardTotal);

        //获取cpu核心数
        int size = Runtime.getRuntime().availableProcessors();

        //查询需要处理的任务
        List<MediaProcess> waitingTaskList = mediaProcessService.getWaitingTask(shardIndex, shardTotal, size);

        log.info("查询到{}个任务", waitingTaskList.size());
        if (waitingTaskList.size() <= 0) {
            return;
        }

        //创建线程池 一个视频一个线程
        ExecutorService executorService = Executors.newFixedThreadPool(size);

        //计数器  它允许一个或多个线程一直等待，直到其他线程执行的一组操作完成
        CountDownLatch countDownLatch = new CountDownLatch(waitingTaskList.size());

        //将处理任务添加进线程池中
        waitingTaskList.forEach(mediaProcess -> {
            executorService.execute(() -> {
                //任务id
                Long id = mediaProcess.getId();
                //抢占任务
                boolean b = mediaProcessService.startTask(id);
                if (!b) {
                    return;
                }

                log.debug("开始执行任务:{}", mediaProcess);

                //下边是处理逻辑
                //桶
                String bucket = mediaProcess.getBucket();
                //存储路径
                String filePath = mediaProcess.getFilePath();
                //原始视频md5值
                String fileId = mediaProcess.getFileId();
                //原始视频名称
                String filename = mediaProcess.getFilename();

                File originFile = null;
                try {
                    originFile = mediaFilesService.downLoadFile(bucket, filePath);
                    if (originFile == null) {
                        log.error("下载视频文件失败{}！", mediaProcess.getBucket().concat(filePath));
                        mediaProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "下载待处理文件失败");
                    }
                } catch (IOException e) {
                    mediaProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "下载待处理文件失败");
                }

                //处理下载的视频文件
                File mp4File = null;
                try {
                    mp4File = File.createTempFile("mp4", ".mp4");
                } catch (IOException e) {
                    log.error("创建mp4临时文件失败");
                    mediaProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "创建mp4临时文件失败");
                    return;
                }
                //视频处理结果
                String result = "";

                try {
                    //开始处理视频
                    Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpegpath, originFile.getAbsolutePath(), mp4File.getAbsolutePath());
                    //开始视频转换，成功将返回success
                    result = videoUtil.generateMp4();
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("处理视频文件:{},出错:{}", mediaProcess.getFilePath(), e.getMessage());
                }
                if (!result.equals("success")) {
                    //记录错误信息
                    log.error("处理视频失败,视频地址:{},错误信息:{}", bucket + filePath, result);
                    mediaProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, result);
                    return;
                }

                //将mp4上传至minio
                //mp4在minio的存储路径
                String objectName = getFilePath(fileId, ".mp4");
                //访问url
                String url = "/" + bucket + "/" + objectName;
                try {
                    mediaFilesService.addMediaFilesToMinIO(mp4File.getAbsolutePath(), "video/mp4", bucket, objectName);
                    //将url存储至数据，并更新状态为成功，并将待处理视频记录删除存入历史
                    mediaProcessService.saveProcessFinishStatus(mediaProcess.getId(), "2", fileId, url, null);
                } catch (Exception e) {
                    log.error("上传视频失败或入库失败,视频地址:{},错误信息:{}", bucket + objectName, e.getMessage());
                    //最终还是失败了
                    mediaProcessService.saveProcessFinishStatus(mediaProcess.getId(), "3", fileId, null, "处理后视频上传或入库失败");
                } finally {
                    countDownLatch.countDown();
                }
            });
        });

        //阻塞，指定最大限制的等待时间，阻塞最多等待一定的时间后就解除阻塞
        countDownLatch.await(30, TimeUnit.MINUTES);
    }

    private String getFilePath(String fileMd5,String fileExt){
        return   fileMd5.substring(0,1) + "/" + fileMd5.substring(1,2) + "/" + fileMd5 + "/" +fileMd5 +fileExt;
    }
}
